package cn.itcast.sink;

import cn.itcast.util.DbUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;

import java.sql.Connection;
import java.sql.PreparedStatement;

/**
 * 写入Mysql,通用对象
 */
public class SinkMysql extends RichSinkFunction<Row> {

    /**
     * 开发步骤：
     * 1.创建构造方法，入参：插入sql
     * 2.初始化mysql
     * 获取JDBC连接
     * 3.封装预处理对象
     * 执行更新操作
     * 4.关流
     */
    //1.创建构造方法，入参：插入sql
    String sql;

    public SinkMysql(String sql) {
        this.sql = sql;
    }

    Connection conn = null;
    PreparedStatement pst = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //2.初始化mysql
        //获取JDBC连接
        conn = DbUtil.getConnByJdbc();
        pst = conn.prepareStatement(sql);
    }

    @Override
    public void invoke(Row value, Context context) throws Exception {
        //执行插入
        for (int i = 0; i < value.getArity(); i++) {
            pst.setObject(i+1,value.getField(i));
        }
        pst.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        if (pst != null) {
            pst.close();
        }
        if (conn != null) {
            conn.close();
        }
    }
}
